package com.example.rabbitmq.publisher.confirms;

import com.example.rabbitmq.constant.Constants;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.Collection;
import java.util.Collections;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.TimeoutException;

public class PublisherCOnfirms {

    static final Integer MESSAGE_COUNT = 3000;
    static final Integer BATCH_SIZE = 50;

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        //单独确认
//        publishingMessageIndividually();
        //批量确认
        publishingMessagesInBatches();
        //异步确认
        publishingMessagesAsynchronously();
    }

    /**
     * 异步确认
     */
    private static void publishingMessagesAsynchronously() throws IOException, TimeoutException, InterruptedException {
        Connection connection=createConnection();
        //创建通道
        Channel channel=connection.createChannel();
        //开启发布确认
        channel.confirmSelect();
        //声明信道
        channel.queueDeclare(Constants.CONFIRM_QUEUE3, true, false, false, null);
        SortedSet<Long> unConfirmSet = Collections.synchronizedSortedSet(new TreeSet<Long>());
        channel.addConfirmListener(new ConfirmListener() {
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                if (multiple) {
                    unConfirmSet.headSet(deliveryTag + 1).clear();
                } else {
                    unConfirmSet.remove(deliveryTag);
                }
            }
            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                if (multiple) {
                    unConfirmSet.headSet(deliveryTag + 1).clear();
                }else {
                    unConfirmSet.remove(deliveryTag);
                }
                System.out.println("未确认消息"+unConfirmSet);
            }
        });
        Long startTime=System.currentTimeMillis();
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = "消息" + i;
            channel.basicPublish("", Constants.CONFIRM_QUEUE3, null, message.getBytes());
            unConfirmSet.add(channel.getNextPublishSeqNo());
        }
//        while (!unConfirmSet.isEmpty()){
//            Thread.sleep(10);
//        }
        Long endTime=System.currentTimeMillis();
        System.out.println("发送"+MESSAGE_COUNT+"条消息耗时"+(endTime-startTime)+"ms");
    }

    /**
     * 批量确认
     */
    private static void publishingMessagesInBatches() throws IOException, TimeoutException, InterruptedException {
        Connection connection=createConnection();
        //创建通道
        Channel channel=connection.createChannel();
        //开启发布确认
        channel.confirmSelect();
        //声明信道
        channel.queueDeclare(Constants.CONFIRM_QUEUE2, true, false, false, null);
        Long startTime=System.currentTimeMillis();
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = "消息" + i;
            channel.basicPublish("", Constants.CONFIRM_QUEUE2, null, message.getBytes());
            if (i % BATCH_SIZE == 0) {
                boolean flag = channel.waitForConfirms();
                if (!flag)
                    throw new RuntimeException("消息发送失败");
            }
        }
        channel.waitForConfirms();
        Long endTime=System.currentTimeMillis();
        System.out.println("发送"+MESSAGE_COUNT+"条消息耗时"+(endTime-startTime)+"ms");
    }

    /**
     * 单独确认
     */
    private static void publishingMessageIndividually() throws IOException, TimeoutException, InterruptedException {
        Connection connection=createConnection();
        //创建通道
        Channel channel=connection.createChannel();
        //开启发布确认
        channel.confirmSelect();
        //声明信道
        channel.queueDeclare(Constants.CONFIRM_QUEUE1, true, false, false, null);
        Long startTime=System.currentTimeMillis();
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = "消息" + i;
            channel.basicPublish("", Constants.CONFIRM_QUEUE1, null, message.getBytes());
            //单独确认
            boolean flag = channel.waitForConfirms();
            if (!flag) {
                throw new RuntimeException("消息发送失败");
            }
        }
        Long endTime=System.currentTimeMillis();
        System.out.println("发送"+MESSAGE_COUNT+"条消息耗时"+(endTime-startTime)+"ms");
    }

    static Connection createConnection() throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(Constants.HOST);
        factory.setPort(Constants.PORT);
        factory.setUsername(Constants.USERNAME);
        factory.setPassword(Constants.PASSWORD);
        factory.setVirtualHost(Constants.VIRTUALHOST);
        //创建连接
        Connection connection = factory.newConnection();
        return connection;
    }
}
